package com.motu.monstercity.server.gate;


import com.motu.monstercity.module.common.network.ResponseUtils;
import com.motu.monstercity.module.common.utility.EventBusConstant;
import com.motu.monstercity.protocol.CmdEnum;
import com.motu.vertx.module.utility.base.ServerConfig;
import com.motu.vertx.module.utility.monitor.MonitorManager;
import com.motu.vertx.module.utility.toolset.ErrorWord;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.core.json.JsonObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;

public class ReceiveVerticle extends AbstractVerticle {
    public final static Logger logger = LoggerFactory.getLogger(ReceiveVerticle.class);

    //	  private Counter activeUserCounter;
    //	  UserService userService;
    //	  private  ConcurrentHashMap<Integer, ServerWebSocket> socketList = new ConcurrentHashMap<Integer, ServerWebSocket>();
    @Override
    public void start(Promise<Void> future) throws Exception {

        EventBus eb = vertx.eventBus();
        String ebName = EventBusConstant.getAddress(EventBusConstant.GATE_RECEIVE) + "_" + ServerConfig.getGateName();
        MessageConsumer<JsonObject> consumer = eb.consumer(ebName);
//		logger.info("---===#*#*#*#===---* gate ReceiveVerticle start  ebName:"+ebName);
        consumer.handler(message -> {

//			long sourceUserId = Long.parseLong(message.headers().get("suid"));
            JsonObject recData = message.body();
            long userId = recData.getLong("userid");
            if (recData.containsKey("kickcode")) {
                // 把用户踢下线
                SocketManager.kickUser(userId, ErrorWord.getEnum(recData.getInteger("kickcode")));
            } else {
                int resCode = recData.getInteger("code");
                byte[] resData = recData.getBinary("data");
			//logger.info("---===#*#*#*#===---*gate Received   cmd:"+SocketManager.getCmdName(resCode)+" userId:" + userId);
                MonitorManager.addProtocolMonitorData(userId, 0, resCode, new byte[0], resData);// 有配置监控时，记录下发的数据内容
                ConcurrentHashMap<Long, ServerWebSocket> list = SocketManager.getSocketList();
                if (list.containsKey(userId)) {
                    ServerWebSocket info = list.get(userId);
					//logger.info("---===#*#*#*#===---*gate send to user:"+userId+"  cmd:"+SocketManager.getCmdName(resCode)+" resData:" + resData.length);
                    ResponseUtils.push(info, resCode, resData);
                }
            }
        });

        consumer.completionHandler(res2 -> {

            if (res2.succeeded()) {

                future.complete();
                logger.info("The handler registration has reached all nodes");

            } else {
                future.fail("The handler registration has failed");
                logger.error("Registration failed! " + res2.cause().toString());

            }

        });

    }


}
